package org.redis.cluster.loadbalance.routing;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.RandomUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.cache.CacheFactory;
import org.redis.cluster.cache.Cache;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.extension.ExtensionLoader;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.support.RedisSlaveProcessor;
import org.redis.cluster.registry.Registry;
import org.redis.cluster.registry.Node;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RoutingManager {
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private volatile Map<ClientConfig,List<ClientConfig>> routingCache=new ConcurrentHashMap<ClientConfig,List<ClientConfig>>();
	
	private RedisCluster redisCluster;
	private ExtensionLoader extensionLoader;
	private Registry registry;
	private Cache cache;
	
	private boolean isEnable;
	
	public RoutingManager() {
		
	}
	
	public RoutingManager(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.extensionLoader=redisCluster.getExtensionLoader();
		this.registry=redisCluster.getRegistry();
		this.cache=initCache();
		this.isEnable=initEnable();
		
		initClientListener();
		initRoutingCache();
	}
	
	protected Cache initCache() {
		return CacheFactory.getCache();
	}
	
	protected boolean initEnable() {
		return RoutingHelper.getEnable(redisCluster);
	}
	
	protected void initClientListener() {
		registry.listenClient((String path)->{
			if(logger.isInfoEnabled()) {
				logger.info("监听到后端redis变化,准备重置路由");
			}
			
			initRoutingCache();
		});
	}
	
	protected boolean initRoutingCache() {
		this.routingCache=getRouting();
		return true;
	}
	
	/**
	 * @return
	 */
	public Map<ClientConfig,List<ClientConfig>> getRouting() {
		Collection<ClientConfig> clients=registry.getClients();
		if(CollectionUtils.isEmpty(clients)) {
			return MapUtils.EMPTY_MAP;
		}
		
		Map<ClientConfig,List<ClientConfig>> routingCache=new ConcurrentHashMap<ClientConfig,List<ClientConfig>>();
		for(ClientConfig client:clients) {
			Collection<Node> routingNodes=registry.getNodes(registry.getClientPath(client), ClientConfig.class);
			if(CollectionUtils.isEmpty(routingNodes)) {
				continue;
			}
			
			List<ClientConfig> routingConfigs=new ArrayList<ClientConfig>(routingNodes.size());
			for(Node routingNode:routingNodes) {
				routingConfigs.add(routingNode.getData());
			}
			
			routingCache.put(client, routingConfigs);
		}
		
		return routingCache;
	}
	
	/**
	 * @param source
	 * @param routing
	 * @return
	 */
	public boolean registerRouting(ClientConfig source,ClientConfig routing) {
		if(!RedisSlaveProcessor.getInstance().slave(routing, source, false)) {
			return false;
		}
		
		return registry.register(registry.getClientPath(source), routing, null);
	}
	
	/**
	 * @param source
	 * @param routing
	 * @return
	 */
	public boolean unregisterRouting(ClientConfig source,ClientConfig routing) {
		RedisSlaveProcessor.getInstance().slaveofNoOne(routing);
		return registry.unregister(registry.getClientPath(source), routing);
	}
	
	/**
	 * @param source
	 * @param routingId
	 * @return
	 */
	public ClientConfig getRouting(ClientConfig source,String routingId) {
		Collection<Node> routingNodes=registry.getNodes(registry.getClientPath(), source, ClientConfig.class);
		if(CollectionUtils.isEmpty(routingNodes)) {
			return null;
		}
		
		for(Node routingNode:routingNodes) {
			ClientConfig config=routingNode.getData();
			if(config.getId().equals(routingId)) {
				return config;
			}
		}
		
		return null;
	}
	
	/**
	 * @param source
	 * @param command
	 * @return
	 */
	public ClientConfig getRouting(ClientConfig source,RedisCommand command) {
		if(!isEnable) {
			return source;
		}
		
		List<ClientConfig> routings=getRouting().get(source);
		if(CollectionUtils.isEmpty(routings)) {
			return source;
		}
		
		if(!isSelectRouting(command)) {
			return source;
		}
		
		return doSelectRouting(routings);
	}
	
	protected boolean isSelectRouting(RedisCommand command) {
		if(!(command.getParameters().length>1)) {
			return false;
		}
		
		return !cache.isExist(command.getParameters()[1]);
	}
	
	protected ClientConfig doSelectRouting(List<ClientConfig> routings) {
		int size=routings.size();
		if(size==1) {
			return routings.get(0);
		}
		
		return routings.get(RandomUtils.nextInt(0, size));
	}
	
	public void setCache(Cache cache) {
		this.cache = cache;
	}
	public Cache getCache() {
		return cache;
	}
	public void setRegistry(Registry registry) {
		this.registry = registry;
	}
	public Registry getRegistry() {
		return registry;
	}
	public void setRedisCluster(RedisCluster redisCluster) {
		this.redisCluster = redisCluster;
	}
	public RedisCluster getRedisCluster() {
		return redisCluster;
	}
}